package bigdata;

import java.sql.Date;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Locale;
import java.util.Map;

import twitter4j.internal.org.json.JSONArray;
import twitter4j.internal.org.json.JSONException;
import twitter4j.internal.org.json.JSONObject;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;


/*
 * the FilterBolt class receives a Json format string and extracts the important information from it into a 
 * newVisit class 
 */

public class FilterBolt extends BaseRichBolt {
	OutputCollector _collector;

	@Override
	public void prepare(Map conf, TopologyContext context,
			OutputCollector collector) {
		_collector = collector;
	}

	@Override
	public void execute(Tuple tuple) {

		if (tuple.getString(0).equals("end")) {

			RunStorm.cluster.killTopology("test");
			RunStorm.cluster.shutdown();
		}

		JSONArray mJsonArray;
		try {

			VisitData newVisit=null;
			String str = tuple.getString(0);

			JSONObject mJsonObject = new JSONObject(str);
			
			JSONObject locationObject = mJsonObject.getJSONObject("location");

			try {
				newVisit = new VisitData(mJsonObject.getLong("aac"),
						mJsonObject.getString("visitId"),
						mJsonObject.getString("ip"),
						mJsonObject.getString("browser"),
						mJsonObject.getString("os"),
						mJsonObject.getString("referrer"),
						mJsonObject.getString("landingPage"),
						new SimpleDateFormat("MMMM d, yyyy HH:mm", Locale.ENGLISH).parse(mJsonObject.getString("onSiteSince")),
						new SimpleDateFormat("MMMM d, yyyy HH:mm", Locale.ENGLISH).parse(mJsonObject.getString("lastIsAlive")),
						mJsonObject.getInt("numberOfVisitedPages"),
						mJsonObject.getBoolean("usedVSBefore"),
						mJsonObject.getBoolean("isFirstVisit"),
						mJsonObject.getInt("numberOfVisitsInSite"),
						mJsonObject.getBoolean("invited"),
						mJsonObject.getBoolean("ignoredInvitaion"),
						mJsonObject.getInt("status"),
						locationObject.getString("countryName"),
						locationObject.getString("continentName"),
						mJsonObject.getLong("timeOnCurrentPage"),
						mJsonObject.getString("searchQuary")
						);
			} catch (ParseException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			SimulationMain.simuFrame.addUserToList(newVisit);
			_collector.emit(tuple, new Values(newVisit));
			_collector.ack(tuple);

		} catch (JSONException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word"));
	}

}
